package com.sirui.x.io.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sirui.x.io.bio.Serializ;

public class SocketChannelWrapper {

	private static final Logger logger = LoggerFactory.getLogger(SocketChannelWrapper.class);
	
	private Selector selector;
	
	public void startAndSend() throws IOException {
		selector = Selector.open();
		SocketChannel sc = SocketChannel.open();
		sc.configureBlocking(false);
		sc.connect(new InetSocketAddress("127.0.0.1", ServerSocketChannelWrapper.LISTEN_PORT));
		sc.register(selector, SelectionKey.OP_CONNECT, "req:a");
		
		// select() 为阻塞
		while(selector.select() > 0) {
			Iterator<SelectionKey> ite = selector.selectedKeys().iterator();
			while(ite.hasNext()) {
				SelectionKey key = ite.next();
				ite.remove();
				// 异步线程池处理
				process(key);
			}
		}
	}
	
	private void process(SelectionKey key) {
		try {
			if(key.isConnectable()) {
				doConnect(key);
			}else if(key.isReadable()) {
				doRead(key);
			}else if(key.isWritable()) {
				doWrite(key);
			}else {
				logger.info("unkown op: " + key.readyOps());
			}
		} catch (IOException e) {
			logger.error("", e);
			try {
				key.channel().close();
			} catch (IOException e1) {
			}
		}
		
	}

	private void doWrite(SelectionKey key) throws IOException {
		SocketChannel sc = (SocketChannel)key.channel();
		String msg = (String)key.attachment();
		byte[] bytes = Serializ.serialization(msg);
		ByteBuffer wrap = ByteBuffer.wrap(bytes);
		sc.write(wrap);
		try {
			logger.info("mock 1 sec..");
			Thread.sleep(1000L);
		} catch (InterruptedException e) {
			logger.error("", e);
		}
		
		ByteBuffer wrap2 = ByteBuffer.wrap("asdfasdfasdfadsfa".getBytes(Charset.forName("UTF8")));
		sc.write(wrap2);
		logger.info("write channel:{}. and regist read.", msg);
		sc.register(selector, SelectionKey.OP_READ);
	}

	private void doRead(SelectionKey key) {
		SocketChannel sc = (SocketChannel)key.channel();
		ByteBuffer buff = ByteBuffer.allocate(4096);
		
		try {
			int len = sc.read(buff);
			if(len > 0) {
				buff.flip();
				logger.info("{read:{}}", Serializ.deserialization(buff.array(), buff.position(), buff.limit()));
				sc.register(selector, SelectionKey.OP_READ);
			}else {
				logger.info("{len:{}}", len);
				sc.close();
			}
			//处理业务逻辑，
		} catch (Exception e) {
			logger.error("", e);
			//关闭连接，
		}
	}

	private void doConnect(SelectionKey key) throws IOException {
		SocketChannel sc = (SocketChannel)key.channel();
		logger.info("{connected and regist write.attach:{}}", key.attachment());
		if(sc.isConnectionPending()) {
			logger.info("connection pending:" + sc.isConnectionPending());
			sc.finishConnect();
		}
		sc.register(selector, SelectionKey.OP_WRITE, key.attach(key.attachment()));
	}
	
}
